package com.qingshu.common.util.zookeeper;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

/**
 * ======================================================================================
 * 方法描述：zookeeper分布式锁
 * 逻辑步骤：
 * 实现 AutoCloseable 可自动关闭资源
 * Watcher观察器
 * ======================================================================================
 */
@Slf4j
public class ZookeeperLock implements AutoCloseable, Watcher {

    /**
     * zookeeper客户端对象
     */
    private ZooKeeper zooKeeper;

    /**
     * 当前节点
     */
    private String currentNode;

    /**
     * 通过构造器初始化zookeeper
     */
    public ZookeeperLock() {
        try {
            //参数：链接字符串 ， 会话超时 ， 监听器
            this.zooKeeper = new ZooKeeper("192.168.237.204:2181", 10000, this);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * ======================================================================================
     * 方法描述：提供获取锁的方法: code作为业务相关的编码，比如传入订单ID，就锁住这个订单
     * 逻辑步骤：
     * 1.创建根节点
     * 2.进来一个线程，就为线程在zookeeper的根节点创建一个瞬时有序节点
     * ======================================================================================
     */
    public boolean getLock(String code) {
        try {
            // 1. 创建根节点，如果不存在，stat为null 就创建
            String path = "/" + code;
            Stat exists = zooKeeper.exists(path, false);
            // 创建节点
            if (null == exists) {
                // 节点路径 ，节点值 ，权限：不需要账户密码就能链接 ， 创建模式：顺序临时节点
                zooKeeper.create(path, code.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }

            // 2. 进来一个线程，就为线程在zookeeper创建一个临时有序节点 ， 如： /xx/xx000000001 ；/xx/xx000000002
            // 把当前节点保存为成员变量，后面用来做判断
            currentNode = zooKeeper.create(path + path, code.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // 3. 对所有的子节点进行排序
            List<String> children = zooKeeper.getChildren(path, false);
            Collections.sort(children);

            // 4. 如果排序过的children中的第一个和 currentNode 一致，拿到锁
            String firstNode = children.get(0);
            if (currentNode.endsWith(firstNode)) {
                return true;
            }

            // 5. 如果不是第一个节点，需要监听前一个节点
            // 用一个临时变量记录当前节点的上一个节点
            String previousNode = firstNode;

            for (String node : children) {
                if (currentNode.endsWith(node)) {
                    // 如果当前节点是node节点 ，那么就监听它的上一个节点 ：比如 currentNode 这里是  0003节点 ，那 node就是 0002节点
                    // 第一个参数是监听的节点，第二个参数是是否要监听，zooKeeper在初始化的时候设置好了监听器
                    log.info("监听上一个节点：{}", node);
                    zooKeeper.exists(path + "/" + previousNode, true);
                } else {
                    //把children中的节点复制给上一个节点
                    previousNode = node;
                }
            }
            //代码到这里，节点已经做好监听了，只需要等待，等待上一个节点完成工作后唤醒他
            synchronized (this) {
                //wait会释放锁
                wait();
            }
            //到这里说明被唤醒，说明获取到锁
            log.info("拿到锁：{}", currentNode);
            return true;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }


    /**
     * 释放锁
     */
    @Override
    public void close() {
        //释放锁 ：节点路径， 节点版本号(-1匹配所有版本)
        log.info("释放节点：{}", currentNode);
        if (null != currentNode) {
            try {
                zooKeeper.delete(currentNode, -1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        //当节点被释放，就会到这里被监听到

        if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
            synchronized (this) {
                //唤醒等待的线程
                log.info("当前节点：{}，唤醒", watchedEvent.getPath());
                notify();
            }
        }
    }
}
